package cn.aiweiyi.qingjing.coupon.distribution.service.impl;

import cn.aiweiyi.qingjing.coupon.comm.constant.Constant;
import cn.aiweiyi.qingjing.coupon.comm.constant.CouponStatus;
import cn.aiweiyi.qingjing.coupon.distribution.dao.CouponDao;
import cn.aiweiyi.qingjing.coupon.distribution.entity.Coupon;
import cn.aiweiyi.qingjing.coupon.distribution.service.KafkaService;
import cn.aiweiyi.qingjing.coupon.distribution.vo.CouponKafkaMessage;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.Optional;

/**
 * @author 青衿
 * @Date 2020/11/21 17:38
 * Kafka相关业务方法,核心思想是,把Redis中优惠券相关的数据
 * 异步地同步到数据库中
 **/
@Slf4j
@Service
public class KafkaServiceImpl implements KafkaService {

    @Autowired
    private CouponDao couponDao;

    /**
     * 小飞优惠券Kafka消息,Topic有消息来了之后,Kafka会调用本方法消费这个消息
     *
     * @param record {@link ConsumerRecord}
     */
    @Override
    @KafkaListener(topics = {Constant.TOPIC}, groupId = "ketai-coupon-1")
    public void cousumeCouponKafkaMessage(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            CouponKafkaMessage coupon = JSON.parseObject(
                    message.toString(), CouponKafkaMessage.class);

            log.info("收到COuponKafkaMessage:{}", message.toString());
            CouponStatus status = CouponStatus.of(coupon.getStatus());

            switch (status) {
                case USABLE:
                    //用户领取优惠券时需要现有ID,也就是已经先持久化到数据库
                    //因此,该状态不需要Kafka异步操作

                    break;

                case USED:
                    processUsedCoupons(coupon, status);
                    break;

                case EXPIRED:
                    processUsedCoupons(coupon, status);
                    break;
            }
        }
    }

    /**
     * 根据状态处理优惠券信息
     *
     * @param kafkaMessage
     * @param status
     */
    private void processCouponsByStatus(CouponKafkaMessage kafkaMessage, CouponStatus status) {
        List<Coupon> coupons = couponDao.findAllById(kafkaMessage.getIds());

        if (CollectionUtils.isEmpty(coupons)
                || coupons.size() != kafkaMessage.getIds().size()) {
            log.error("无法找到之前的优惠券信息:{}",
                    JSON.toJSONString(kafkaMessage));

            //发送邮件
            return;
        }

        coupons.forEach(c -> c.setStatus(status));
        log.info("CouponKafkaMessage处理数量:{}",
                couponDao.saveAll(coupons).size());
    }


    /**
     * 处理已使用的用户优惠券
     *
     * @param kafkaMessage
     * @param status
     */
    private void processUsedCoupons(CouponKafkaMessage kafkaMessage, CouponStatus status) {
        //给用户发送短信
        processCouponsByStatus(kafkaMessage, status);
    }

    /**
     * 处理已过期的用户优惠券,该方法目前processUsedCOupons代码逻辑与一致,但以后
     * 业务可能扩展,所有先冗余
     *
     * @param kafkaMessage
     * @param status
     */
    private void processExpiredCoupons(CouponKafkaMessage kafkaMessage, CouponStatus status) {
        //给用户发送推送
        processCouponsByStatus(kafkaMessage, status);
    }
}
